package org.atlantis.datasink.server.buildin;

import java.net.SocketAddress;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;

import org.atlantis.datasink.codec.CodecOption;
import org.atlantis.datasink.codec.ICodecContext;
import org.atlantis.datasink.server.AbstractCodecServer;
import org.atlantis.datasink.server.CommType;
import org.atlantis.datasink.utils.SimpleExecutor;
import org.atlantis.datasink.utils.SubstepAwait;
import org.jboss.netty.bootstrap.Bootstrap;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

public class ActiveTCPServer extends AbstractCodecServer {

	private List<Bootstrap> bootstraps = Collections.synchronizedList(new LinkedList<Bootstrap>());
	private SimpleExecutor executor = new SimpleExecutor();

	@Override
	public CommType getCommType() {
		return CommType.ACTIVE;
	}

	@Override
	protected void startByCodec(ICodecContext codecCtx) {
		final CodecOption codecOpt = codecCtx.getOption();
		SubstepAwait await = SubstepAwait.create(codecOpt.getRequestServers().size());
		for (SocketAddress addr : codecCtx.getOption().getRequestServers()) {
			final SocketAddress _a = addr;
			await.exec(new Runnable() {
				@Override
				public void run() {
					ClientBootstrap bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(executor, executor));
					bootstrap.setOption("remoteAddress", _a);
					bootstrap.setOption("tcpNoDelay", true);
					bootstrap.setOption("codecOption", codecOpt);
					ActiveConnectManager.getInstance().connect(bootstrap);
					bootstraps.add(bootstrap);
				}
			});
		}
		await.waitFinish();
	}

	@Override
	public void shutdown() {
		ActiveConnectManager.getInstance().destory();
		for (Bootstrap bootstrap : bootstraps) {
			bootstrap.releaseExternalResources();
		}
		executor.destory();
	}

	@Override
	public String getName() {
		return "Active-TCP-Server";
	}
}
